package com.example.hadoop;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.util.StringUtils;

import java.io.*;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;
import java.util.StringTokenizer;


public class HeadLineCounter1 {
    public static class WordMapper1 extends Mapper<Object, Text, Text, IntWritable> {
        private final Text word = new Text();
        private final IntWritable one = new IntWritable(1);
        private List<String> patternsToSkip=new ArrayList<>();
        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            Configuration conf = context.getConfiguration();
            URI[] patternsURIs = Job.getInstance(conf).getCacheFiles();
            if (patternsURIs == null) {
                throw new IOException("No cache files found.");
            }
            patternsToSkip = new ArrayList<>();
            for (URI patternsURI : patternsURIs) {
                if (patternsURI == null) {
                    throw new IOException("Pattern URI is null.");
                }
                Path patternsPath = new Path(patternsURI.getPath());
                if (patternsPath == null) {
                    throw new IOException("Pattern path is null.");
                }
                readSkipFile(patternsPath, conf);
            }

        }

        private void readSkipFile(Path path, Configuration conf) throws IOException {
            FileSystem fs = FileSystem.get(conf);
            InputStream in = null;
            try {
                in = fs.open(path);
                BufferedReader reader = new BufferedReader(new InputStreamReader(in));
                String line;
                while ((line = reader.readLine()) != null) {
                    // 可以在这里添加额外的停用词处理逻辑，比如转换为小写或去除空白字符
                    String pattern = line.trim();
                    if (!pattern.isEmpty()) {
                        patternsToSkip.add(pattern);
                    }
                }
            } finally {
                if (in != null) {
                    in.close();
                }
            }
        }

        @Override
        protected void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString().toLowerCase();
            for(String pattern:patternsToSkip){
                line=line.replaceAll(pattern,"");
            }
            StringTokenizer itr=new StringTokenizer(line);
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());
                String s=word.toString();
                String regEx = "[`~☆★!@#$%^&*()+=|{}':;,\\[\\]》·.<>/?~！@#￥%……（）——+|{}【】‘；：”“’。\"，\\-、？]";
                String s1=s.replaceAll(regEx,"");
                Text word2=new Text(s1);
                context.write(word2, one);
            }
        }
    }
    public static class WordReducer1 extends Reducer<Text, IntWritable, Text, IntWritable> {
        private final IntWritable result = new IntWritable();

        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            result.set(sum);
            context.write(key, result);
        }
    }
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "headline counter stage1");

        job.setJarByClass(HeadLineCounter1.class);
        job.setMapperClass(WordMapper1.class);
        job.setReducerClass(WordReducer1.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 设置是否跳过模式匹配，默认为true
        conf.setBoolean("wordcount.skip.patterns", true);
        // 添加停用词文件到DistributedCache
        // 假设停用词文件已经上传到HDFS上的/path/to/stop-word-list.txt
        String stopWordsFile = "hdfs:///path/to/stop-word-list.txt";
        job.addCacheFile(new URI(stopWordsFile));


        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }

}
